package com.myself.sender;

import com.myself.config.ExpirationMessagePostProcessor;
import com.myself.config.SenderQueueConfig;
import com.myself.model.User;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;

/**
 * @author W.S.
 * @version v4.0.0
 * @description
 * @date Created in 2018/8/21 17:05
 */
@Service
public class SenderService implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback,InitializingBean{

    private static Logger logger = LoggerFactory.getLogger(SenderService.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendRabbitMsg(String str){
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        logger.info("发送 rabbit MQ 消息");
        Message message = MessageBuilder.withBody(str.getBytes()).build();
        rabbitTemplate.convertAndSend("direct.exchange","test1",message,correlationData);
    }

    public void sendUser(User user){
        logger.info("发送 rabbit MQ 消息 user:{}",user.toString());
        rabbitTemplate.convertAndSend("direct.exchange"," ",user);
    }

    /**
     * 发送topic 消息
     * @param str
     * @param routingKey
     */
    public void sendTopicStr(String str,String routingKey){
        logger.info("发送 rabbit MQ topic 消息 str:{}",str);
        rabbitTemplate.convertAndSend("topic.exchange",routingKey,str);
    }

    /**
     * 发送延时对列消息
     */
    public void sendDelayMsg(String str,String routingKey){
        logger.info("发送延时队列消息 str:{},routingKey:{}",str,routingKey);
        rabbitTemplate.convertAndSend("DEAD_LETTER_EX",routingKey,str);
    }

    /**
     *
     * @param msg
     * @param routingKey
     * @param expration
     */
    public void sendDelayExprationMsg(String msg,String routingKey,String expration){
        logger.info("发送自定义消息失效时间的msg:{} ,expration:{}",msg,expration);
        rabbitTemplate.convertAndSend("DEAD_LETTER_EX",routingKey,msg,new ExpirationMessagePostProcessor("30000"));
        //amqpTemplate.sendAndReceive("DEAD_LETTER_EX",routingKey,message);
    }

    /**
     * Confirmation callback.
     *
     * @param correlationData correlation data for the callback.
     * @param ack             true for ack, false for nack
     * @param cause           An optional cause, for nack, when available, otherwise null.
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if(ack){
            logger.info("消息发送到exchange成功,id: {}",correlationData.getId());
        }else {
            logger.info("消息发送到exchange失败,原因: {}", cause);
        }
    }

    /**
     * Returned message callback.
     *
     * @param message    the returned message.
     * @param replyCode  the reply code.
     * @param replyText  the reply text.
     * @param exchange   the exchange.
     * @param routingKey the routing key.
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        String correlationId = message.getMessageProperties().getCorrelationId();
        logger.info("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {}  路由键: {}", correlationId, replyCode, replyText, exchange, routingKey);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        rabbitTemplate.setConfirmCallback(this::confirm);
        rabbitTemplate.setReturnCallback(this::returnedMessage);
    }
}
